package com.disruptor.sample._07advance.run;

import com.disruptor.sample._07advance.consumer.TradeHandler;
import com.disruptor.sample._07advance.dto.Trade;
import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.YieldingWaitStrategy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class AdvanceMain01 {

  public static RingBuffer<Trade> buildRingBuffer() {
    int BUFFER_SIZE = 1024;
    /*
     * createSingleProducer 创建一个单生产者的 RingBuffer，
     * 第一个参数叫 EventFactory，从名字上理解就是"事件工厂"，其实它的职责就是产生数据填充 RingBuffer 的区块。
     * 第二个参数是 RingBuffer 的大小，它必须是 2 的指数倍 目的是为了将求模运算转为 & 运算提高效率
     * 第三个参数是 RingBuffer 的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
     */
    final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<Trade>() {
      @Override
      public Trade newInstance() {
        return new Trade();
      }
    }, BUFFER_SIZE, new YieldingWaitStrategy());

    return ringBuffer;
  }

  public static ExecutorService buildProcessExecutors() {
    // 创建线程池
    int THREAD_NUMBERS = 4;
    ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);
    return executors;
  }

  public static List<BatchEventProcessor<Trade>> buildProcessors(ExecutorService executors,
      final RingBuffer<Trade> ringBuffer, SequenceBarrier sequenceBarrier, int num) {
    List<BatchEventProcessor<Trade>> processors = new ArrayList<>();
    for (int i = 0; i < num; i++) {
      // 多个消费者 那重复执行以下代码
      // 创建消息处理器，一个生产者，对应多个消费者
      BatchEventProcessor<Trade> transProcessor = new BatchEventProcessor<Trade>(
          ringBuffer, sequenceBarrier, new TradeHandler());

      // 这一步的目的就是把消费者的位置信息引用注入到生产者，如果只有一个消费者的情况可以省略 ，将生产者和消费者关联起来
      ringBuffer.addGatingSequences(transProcessor.getSequence());

      // 把消息处理器提交到线程池
      executors.submit(transProcessor);

      processors.add(transProcessor);
    }
    return processors;
  }

  public static void main(String[] args) throws Exception {
    final RingBuffer<Trade> ringBuffer = buildRingBuffer();

    ExecutorService executors = buildProcessExecutors();

    // 创建 SequenceBarrier ,用于平衡生产者和消费者速率，用障碍来处理
    SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
    List<BatchEventProcessor<Trade>> processors = buildProcessors(executors, ringBuffer,
        sequenceBarrier, 3);

    Future<?> future = executors.submit(new Callable<Void>() {
      @Override
      public Void call() throws Exception {
        long seq;
        for (int i = 0; i < 5; i++) {
          // 占个坑 --ringBuffer 一个可用区块
          seq = ringBuffer.next();
          // 给这个区块放入 数据
          ringBuffer.get(seq).setPrice(Math.random() * 9999);
          // 发布这个区块的数据使 handler(consumer) 可见
          ringBuffer.publish(seq);
        }
        return null;
      }
    });

    // 等待生产者结束
    future.get();
    // 等消费都处理完成
    Thread.sleep(5000);
    // 通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）
    for (BatchEventProcessor<Trade> transProcessor : processors) {
      transProcessor.halt();
    }
    // 终止线程
    executors.shutdown();
  }
}
